package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.IndexBean;
import cn.itcast.bean.StockBean;
import cn.itcast.function.IndexMinWindowFunction;
import cn.itcast.function.KeyFunction;
import cn.itcast.function.StockMinWindowFunction;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.util.DorisUtil;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 指数分时行情
 */
public class IndexMinTask implements ProcessDataInterface {
    @Override
    public void process(DataStream<CleanBean> waterData) {
        /**
         * 开发步骤：
         * 1.定义侧边流
         * 2.数据分组
         * 3.划分时间窗口
         * 4.分时数据处理（新建分时窗口函数）
         * 5.数据分流
         * 6.数据分流转换
         * 7.分表存储(Doris)
         */
        //1.定义侧边流,分流数据->分表存储
        //封装深市个股数据
        OutputTag<IndexBean> szseOpt = new OutputTag<>("szseOpt", TypeInformation.of(IndexBean.class));

        //2.数据分组
        SingleOutputStreamOperator<IndexBean> processData = waterData.keyBy(new KeyFunction())
                //3.划分时间窗口(60)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //4.分时数据处理（新建分时窗口函数）
                .apply(new IndexMinWindowFunction())
                //5.数据分流
                .process(new ProcessFunction<IndexBean, IndexBean>() {
                    @Override
                    public void processElement(IndexBean value, Context ctx, Collector<IndexBean> out) throws Exception {
                        if (value.getSource().equals("sse")) {
                            out.collect(value);
                        } else {
                            ctx.output(szseOpt, value);
                        }
                    }
                });
        //6.分表存储(写入Doris)
        //sse -> str
        processData.addSink(DorisUtil.getSinkFunction("insert into quot_index_sse values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

        //szse - > str
        processData.getSideOutput(szseOpt).print("<<<<<szse<<<<:");
        processData.getSideOutput(szseOpt).addSink(DorisUtil.getSinkFunction("insert into quot_stock_szse values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

    }
}
